package org.obm.cassandra.client.tests;

import static me.prettyprint.cassandra.utils.StringUtils.string;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import me.prettyprint.cassandra.service.Keyspace;

import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.obm.bigg.model.Attachments;
import org.obm.cassandra.client.CassHelper;

public class HectorTests extends CassandraTest {

	private static final int COUNT = 5000;

	public void testConnection() {
		assertNotNull(client);
	}

	public void testInsertAndRead() {
		try {
			Keyspace keyspace = client.getKeyspace("bigg");
			String key = "attach_" + System.currentTimeMillis();

			Map<String, String> data = new HashMap<String, String>();
			data.put("filename", "toto.pdf");
			data.put("content", "base64content");
			data.put("size", "561");
			CassHelper.insert(keyspace, Attachments.CF, key, data);

			// read
			ColumnPath cp = CassHelper.cp(Attachments.CF, "filename");
			Column col = keyspace.getColumn(key, cp);
			System.out
					.println("Read from cassandra: " + string(col.getValue()));

			String[] keys = { key };
			List<String> list = Arrays.asList(keys);
			Map<String, Column> values = keyspace.multigetColumn(list, cp);
			for (String s : values.keySet()) {
				System.out.println(" - " + s + ": "
						+ string(values.get(s).getValue()));
			}

			System.err.println("trying to fetch all columns");
			ColumnParent columnParent = new ColumnParent(Attachments.CF);
			SliceRange sr = new SliceRange(new byte[0], new byte[0], false, 100);
			SlicePredicate predicate = new SlicePredicate();
			predicate.setSlice_range(sr);
			List<Column> results = keyspace.getSlice(key, columnParent,
					predicate);
			System.err.println("results length: " + results.size());
			for (Column c : results) {
				System.err.println(" - " + string(c.getName()) + " --> "
						+ string(c.getValue()));
			}

		} catch (Exception e) {
			e.printStackTrace();
			fail();
		}
	}

	public void testReadPerf() {
		try {
			Keyspace keyspace = client.getKeyspace("bigg");

			String key = "attach_" + System.currentTimeMillis();

			Map<String, String> data = new HashMap<String, String>();
			data.put("filename", "toto.pdf");
			data.put("content", "base64content");
			data.put("size", "561");
			CassHelper.insert(keyspace, "Attachments", key, data);

			// read one time
			ColumnPath columnPath = CassHelper.cp(Attachments.CF, "filename");
			Column value = keyspace.getColumn(key, columnPath);
			System.err.println("value: " + string(value.getName()) + " "
					+ string(value.getValue()));

			// warm up JIT
			for (int i = 0; i < COUNT; i++) {
				for (String s : Attachments.cols) {
					ColumnPath cp = CassHelper.cp(Attachments.CF, s);
					keyspace.getColumn(key, cp);
				}
			}

			long time = System.currentTimeMillis();
			for (int i = 0; i < COUNT; i++) {
				for (String s : Attachments.cols) {
					ColumnPath cp = CassHelper.cp(Attachments.CF, s);
					keyspace.getColumn(key, cp);
				}
			}
			time = System.currentTimeMillis() - time;
			System.out.println("read all " + Attachments.cols.size()
					+ " fields " + COUNT + "times in " + time + "ms. "
					+ (COUNT * 1000) / time + " per sec.");

		} catch (Exception e) {
			e.printStackTrace();
			fail();
		}
	}
}
